package org.raymond.iworks.core.serializer;

import lombok.extern.slf4j.Slf4j;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class KafkaAvroSerializer<T extends SpecificRecordBase> implements Serializer<T> {
    @Override
    public byte[] serialize(String topic, T data) {
        byte[] bytes = null;
        try (ByteArrayOutputStream os = new ByteArrayOutputStream()){
            DatumWriter<T> writer = new SpecificDatumWriter<>(data.getSchema());
            BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(os, null);
            writer.write(data, binaryEncoder);
            binaryEncoder.flush();
            bytes = os.toByteArray();
        }catch (Exception e){
            log.error("{}", e.getMessage());
        }
        return bytes;
    }
}
